package org.apache.rocketmq.store;



import org.apache.rocketmq.common.ServiceThread;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.store.config.BrokerRole;
import org.apache.rocketmq.store.logfile.DefaultMappedFile;
import org.apache.rocketmq.store.logfile.MappedFile;

import java.io.File;
import java.io.IOException;
import java.util.ServiceLoader;
import java.util.concurrent.*;

/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/6/29
 * @方法描述：内存映射文件分配器
 */
public class AllocateMappedFileService extends ServiceThread {

    private static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);

    //内存映射文件等待分配的超时时间，默认为5秒
    //如果超过5秒还没有成功分配内存分配文件，那就意味着分配失败了
    private static int waitTimeOut = 1000 * 5;

    //存放分配内存映射文件请求的map，key为要分配的内存映射文件的路径，value就是对应的分配请求对象
    private ConcurrentMap<String, AllocateRequest> requestTable = new ConcurrentHashMap<>();

    //存放分配内存映射文件请求的队列，当前映射文件分配器其实也是一个后台线程，启动之后会在run方法中循环从该队列中获取分配请求对象，并执行分配操作
    private PriorityBlockingQueue<AllocateRequest> requestQueue = new PriorityBlockingQueue<>();

    //分配过程中是否发生异常的标志
    private volatile boolean hasException = false;

    private DefaultMessageStore messageStore;

    //构造方法
    public AllocateMappedFileService(DefaultMessageStore messageStore) {
        this.messageStore = messageStore;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：提交内存映射文件分配请求的方法，该方法会被MappedFileQueue对象调用
     */
    public MappedFile putRequestAndReturnMappedFile(String nextFilePath, String nextNextFilePath, int fileSize) {
        //可提交的请求数量，默认为2，因为在mq创建内存映射文件时，一次会创建两个文件，把本次要用的和下一个要用的文件都创建出来
        //提交两个请求就意味着要创建两个对应的内存映射文件，以此来提高消息存储效率
        int canSubmitRequests = 2;
        //判断是否启用了临时存储池功能，启用了临时存储池就实现了内存级别的读写分离
        if (this.messageStore.isTransientStorePoolEnable()) {
            //判断程序是否配置了在存储池没有足够缓冲区时快速失败的标志，同时也要判断一下当前节点不是从节点
            if (this.messageStore.getMessageStoreConfig().isFastFailIfNoBufferInStorePool() && BrokerRole.SLAVE != this.messageStore.getMessageStoreConfig().getBrokerRole()) {
                //在这里更新本次可提交的请求的数量，计算可以提交的请求数量，其实就是剩余缓冲区数量减去当前请求队列大小
                //这确保了要提交的请求数量不会超过可用的临时存储缓冲区数量，因为如果启用了临时存储池，那么每创建一个内存映射文件，就会从临池存储池中分配一个堆外缓冲区给内存映射文件
                canSubmitRequests = this.messageStore.remainTransientStoreBufferNumbs() - this.requestQueue.size();
            }
        }
        //为要创建的内存映射文件创建对应的请求对象，nextFilePath是要创建的文件路径，fileSize是文件大小
        //这里我多说一句，大家肯定也注意到了，在当前方法的参数中可以看到第二个参数为nextNextFilePath，这个参数就是要创建的下下个文件路径，后面就会用到了
        AllocateRequest nextReq = new AllocateRequest(nextFilePath, fileSize);
        //在这里把请求对象和路径放进请求表中
        boolean nextPutOK = this.requestTable.putIfAbsent(nextFilePath, nextReq) == null;
        if (nextPutOK) {
            //如果请求存放成功，对要提交的请求数量进行合法校验
            if (canSubmitRequests <= 0) {
                //小于0则记录警告日志，表明临时存储池缓冲区不足
                log.warn("[NOTIFYME]TransientStorePool is not enough, so create mapped file error, " +
                        "RequestQueueSize : {}, StorePoolSize: {}", this.requestQueue.size(), this.messageStore.remainTransientStoreBufferNumbs());
                //然后从请求表中移除刚才添加的对应请求
                this.requestTable.remove(nextFilePath);
                //返回null，表示无法分配文件
                return null;
            }
            //接下来就要把请求对象添加到请求队列中了
            boolean offerOK = this.requestQueue.offer(nextReq);
            //判断是否成功添加了
            if (!offerOK) {
                //这里的日志信息也解释了，理论上来说，程序是不会执行到这里的，也就是说肯定会添加成功
                //除非系统资源极度紧张
                log.warn("never expected here, add a request to preallocate queue failed");
            }
            //执行到这里意味着本次要创建的文件的请求已经提交成功了，这个时候要减少可提交请求数量
            //这就意味着要开始提交下一个内存映射文件创建的请求了
            canSubmitRequests--;
        }
        //以下就是创建下下个内存映射文件分配请求的操作
        //创建下下个文件的分配请求对象，这是实现预分配机制的关键，提前分配下一个文件，减少消息写入时的等待时间
        AllocateRequest nextNextReq = new AllocateRequest(nextNextFilePath, fileSize);
        //在这里把请求对象和路径放进请求表中
        boolean nextNextPutOK = this.requestTable.putIfAbsent(nextNextFilePath, nextNextReq) == null;
        //以下的操作判断和刚才的操作一样，我就不添加注释了
        if (nextNextPutOK) {
            if (canSubmitRequests <= 0) {
                log.warn("[NOTIFYME]TransientStorePool is not enough, so skip preallocate mapped file, " +
                        "RequestQueueSize : {}, StorePoolSize: {}", this.requestQueue.size(), this.messageStore.remainTransientStoreBufferNumbs());

                this.requestTable.remove(nextNextFilePath);
            } else {
                boolean offerOK = this.requestQueue.offer(nextNextReq);
                if (!offerOK) {
                    log.warn("never expected here, add a request to preallocate queue failed");
                }
            }
        }
        //判断操作执行过程中是否有异常
        //如果有异常，则直接返回null，表示分配失败
        if (hasException) {
            log.warn(this.getServiceName() + " service has exception. so return null");
            return null;
        }

        //接下来就是等到内存映射文件创建成功的操作了
        //从请求表中获取当前要创建的内存映射文件的分配请求对象
        AllocateRequest result = this.requestTable.get(nextFilePath);
        try {
            //如果找到了当前文件的分配请求
            if (result != null) {
                //就在这里等待映射文件分配完成
                //使用CountDownLatch实现线程间的等待通知机制，等待时间为waitTimeOut毫秒，超时返回false
                //注意，这里阻塞的线程是调用当前putRequestAndReturnMappedFile()方法的线程，这个要搞清楚
                boolean waitOK = result.getCountDownLatch().await(waitTimeOut, TimeUnit.MILLISECONDS);
                //如果等待超时
                if (!waitOK) {
                    //则记录警告日志，表明创建内存映射文件超时
                    log.warn("create mmap timeout " + result.getFilePath() + " " + result.getFileSize());
                    //返回null，表示分配失败
                    return null;
                } else {
                    //执行到这里意味着没有超时，而是创建内存映射文件成功之后被唤醒了
                    //这个时候就可以把请求对象从表中移除了
                    this.requestTable.remove(nextFilePath);
                    //返回成功分配的MappedFile对象
                    return result.getMappedFile();
                }
            } else {
                //如果没有从表中找到对应的内存映射文件创建请求，则记录错误日志
                //从日志信息中也可以看到，这种情况实际上也不会发生
                log.error("find preallocate mmap failed, this never happen");
            }
        } catch (InterruptedException e) {
            log.warn(this.getServiceName() + " service has exception. ", e);
        }
        return null;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：获取当前服务组件名称的方法
     */
    @Override
    public String getServiceName() {
        if (messageStore != null && messageStore.getBrokerConfig().isInBrokerContainer()) {
            return messageStore.getBrokerIdentity().getIdentifier() + AllocateMappedFileService.class.getSimpleName();
        }
        return AllocateMappedFileService.class.getSimpleName();
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：终止当前服务组件工作的方法
     */
    @Override
    public void shutdown() {
        super.shutdown(true);
        for (AllocateRequest req : this.requestTable.values()) {
            if (req.mappedFile != null) {
                log.info("delete pre allocated maped file, {}", req.mappedFile.getFileName());
                //销毁内存映射文件，释放系统资源，参数1000表示最多等待1000毫秒完成资源释放
                req.mappedFile.destroy(1000);
            }
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：当前服务组件启动之后要执行的方法
     */
    public void run() {
        log.info(this.getServiceName() + " service started");
        //在循环中执行分配内存映射文件的操作，分配内存映射文件的操作就在mmapOperation()方法中执行
        while (!this.isStopped() && this.mmapOperation()) {

        }
        log.info(this.getServiceName() + " service end");
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：分配内存映射文件的方法
     */
    private boolean mmapOperation() {
        //先定义一个内存映射文件分配是否成功的标志
        boolean isSuccess = false;
        //当前正在处理的内存映射文件分配请求
        AllocateRequest req = null;

        try {
            //从请求队列中阻塞获取一个请求
            req = this.requestQueue.take();
            //根据请求要创建的文件路径从映射表中得到对应的请求对象
            //因为映射表中的请求对象会在文件分配成功后被移除，所以如果这个时候还能得到对应的请求对象
            //就意味着这个请求对应的映射文件确实还没有创建呢，接下来就可以执行创建操作了
            AllocateRequest expectedRequest = this.requestTable.get(req.getFilePath());
            //这里就是具体的判断操作，判断映射表中是否还存在对应的请求
            if (null == expectedRequest) {
                //如果找不到则记录警告日志，表明请求已超时
                log.warn("this mmap request expired, maybe cause timeout " + req.getFilePath() + " " + req.getFileSize());
                //返回true，表示继续处理队列中下一个请求
                return true;
            }
            //如果请求表中的对象与队列中取出的对象不是同一个实例
            if (expectedRequest != req) {
                //记录警告日志，从日志中也可以看出来，其实这种情况理论上也不会发生
                log.warn("never expected here,  maybe cause timeout " + req.getFilePath() + " "
                        + req.getFileSize() + ", req:" + req + ", expectedRequest:" + expectedRequest);
                //返回true，表示继续处理队列中下一个请求
                return true;
            }
            //判断内存映射文件是否已创建完毕，因为创建好的内存映射文件会先交给AllocateRequest请求对象保管
            if (req.getMappedFile() == null) {
                //如果请求对象没有持有内存映射文件，就意味着内存映射文件还没创建，接下来就可以创建了
                //记录创建内存映射文件操作的开始时间
                long beginTime = System.currentTimeMillis();
                MappedFile mappedFile;
                //判断程序是否启用了临时存储池
                if (messageStore.isTransientStorePoolEnable()) {
                    try {
                        //如果启用了则使用SPI机制加载用户自定义的内存映射文件对象
                        mappedFile = ServiceLoader.load(MappedFile.class).iterator().next();
                        //在这里使用临时存储池初始化内存映射文件对象，给该对象分配用来写入数据的堆外内存
                        mappedFile.init(req.getFilePath(), req.getFileSize(), messageStore.getTransientStorePool());
                    } catch (RuntimeException e) {
                        //如果加载自定义内存映射对象失败，记录警告并使用默认实现
                        log.warn("Use default implementation.");
                        //创建默认的内存映射文件对象，也用到了临时存储池
                        mappedFile = new DefaultMappedFile(req.getFilePath(), req.getFileSize(), messageStore.getTransientStorePool());
                    }
                } else {
                    //如果程序没有启用临时存储池功能，则直接创建默认的内存映射文件对象
                    mappedFile = new DefaultMappedFile(req.getFilePath(), req.getFileSize());
                }
                //得到操作执行的耗时
                long elapsedTime = UtilAll.computeElapsedTimeMilliseconds(beginTime);
                //如果操作耗时超过10毫秒，则记录警告日志
                if (elapsedTime > 10) {
                    //获取当前请求队列的大小，用于日志记录，队列大小可以反映系统负载情况
                    int queueSize = this.requestQueue.size();
                    //记录警告日志，包含操作耗时、队列大小、文件路径和大小信息
                    log.warn("create mappedFile spent time(ms) " + elapsedTime + " queue size " + queueSize + " " + req.getFilePath() + " " + req.getFileSize());
                }
                //接下来要执行一个比较重要的操作，也就是对刚创建的内存映射文件进行预热的操作
                //当然，在执行之前要先判断一下程序是否启用了内存文件预热功能
                if (mappedFile.getFileSize() >= this.messageStore.getMessageStoreConfig().getMappedFileSizeCommitLog() && this.messageStore.getMessageStoreConfig().isWarmMapedFileEnable()) {
                    //如果启用了内存文件预热功能，则在这里执行预热操作，我想简单解释一下这里预热操作的作用，实际上是这样的，当内存映射文件创建完毕，这个时候所做的也仅仅是把硬盘文件和虚拟内存的地址映射完毕了
                    //而操作系统实际上并没有为虚拟内存地址分配真正的物理内存呢，当程序需要访问某个虚拟内存的时候，会发现这个虚拟内存并没有对应的物理内存可以使用，这个时候就会触发操作系统缺页中断，然后为这个虚拟地址分配物理内存
                    //当内存映射文件创建成功之后，虽然文件已经和虚拟内存映射了，但是并没有真正的分配物理内存，只有当消息要被第一次写入的时候才会触发页中断然后分配真正的物理内存，这样的话就得等待物理内存分配完毕之后才能写入消息
                    //这显然是比较耗时的操作，需要等待一会，而内存映射文件肯定会映射到多个操作系统缓存页中，如果向每一页第一次写入数据时都触发一次页中断，那消息写入的节奏肯定会被拖慢
                    //所以，这里就对内存映射文件执行了预热操作，具体操作大家可以去mappedFile.warmMappedFile()方法内部查看，其实就是对内存映射文件映射的每一个操作系统缓存页提前写入一个字节0，这样就会提前出发页中断，提前为虚拟内存分配物理空间
                    //等真正写入消息的时候就不会再等待缺页中断，然后再写入消息了，这就是mq的内存映射文件预热功能的原理，在源码中，这个功能默认是没有开启的，实际上临时存储池也没有开启，但这不妨碍大家了解它们的原理，然后在生产中使用这些功能
                    mappedFile.warmMappedFile(this.messageStore.getMessageStoreConfig().getFlushDiskType(), this.messageStore.getMessageStoreConfig().getFlushLeastPagesWhenWarmMapedFile());
                }
                //吧创建完毕的内存映射文件设置到请求对象中
                req.setMappedFile(mappedFile);
                //清除异常标志，表明当前操作没有发生异常
                this.hasException = false;
                //设置操作成功标志
                isSuccess = true;
            }
        } catch (InterruptedException e) {
            log.warn(this.getServiceName() + " interrupted, possibly by shutdown.");
            this.hasException = true;
            return false;
        } catch (IOException e) {
            log.warn(this.getServiceName() + " service has exception. ", e);
            this.hasException = true;
            if (null != req) {
                requestQueue.offer(req);
                try {
                    Thread.sleep(1);
                } catch (InterruptedException ignored) {

                }
            }
        } finally {
            //如果内存映射文件创建成功
            if (req != null && isSuccess) {
                //调用CountDownLatch的countDown方法，唤醒正在putRequestAndReturnMappedFile方法中阻塞的线程
                req.getCountDownLatch().countDown();
            }
        }
        return true;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：分配内存映射文件的请求，这个请求的内容非常简单，我就不添加注释了，大家自己看看就行
     */
    static class AllocateRequest implements Comparable<AllocateRequest> {
        private String filePath;
        private int fileSize;
        private CountDownLatch countDownLatch = new CountDownLatch(1);
        private volatile MappedFile mappedFile = null;

        public AllocateRequest(String filePath, int fileSize) {
            this.filePath = filePath;
            this.fileSize = fileSize;
        }

        public String getFilePath() {
            return filePath;
        }

        public void setFilePath(String filePath) {
            this.filePath = filePath;
        }

        public int getFileSize() {
            return fileSize;
        }

        public void setFileSize(int fileSize) {
            this.fileSize = fileSize;
        }

        public CountDownLatch getCountDownLatch() {
            return countDownLatch;
        }

        public void setCountDownLatch(CountDownLatch countDownLatch) {
            this.countDownLatch = countDownLatch;
        }

        public MappedFile getMappedFile() {
            return mappedFile;
        }

        public void setMappedFile(MappedFile mappedFile) {
            this.mappedFile = mappedFile;
        }

        public int compareTo(AllocateRequest other) {
            if (this.fileSize < other.fileSize)
                return 1;
            else if (this.fileSize > other.fileSize) {
                return -1;
            } else {
                int mIndex = this.filePath.lastIndexOf(File.separator);
                long mName = Long.parseLong(this.filePath.substring(mIndex + 1));
                int oIndex = other.filePath.lastIndexOf(File.separator);
                long oName = Long.parseLong(other.filePath.substring(oIndex + 1));
                if (mName < oName) {
                    return -1;
                } else if (mName > oName) {
                    return 1;
                } else {
                    return 0;
                }
            }
            // return this.fileSize < other.fileSize ? 1 : this.fileSize >
            // other.fileSize ? -1 : 0;
        }

        @Override
        public int hashCode() {
            final int prime = 31;
            int result = 1;
            result = prime * result + ((filePath == null) ? 0 : filePath.hashCode());
            result = prime * result + fileSize;
            return result;
        }

        @Override
        public boolean equals(Object obj) {
            if (this == obj)
                return true;
            if (obj == null)
                return false;
            if (getClass() != obj.getClass())
                return false;
            AllocateRequest other = (AllocateRequest) obj;
            if (filePath == null) {
                if (other.filePath != null)
                    return false;
            } else if (!filePath.equals(other.filePath))
                return false;
            if (fileSize != other.fileSize)
                return false;
            return true;
        }
    }
}
